use std::collections::BTreeMap;
use std::sync::Arc;

use crate::abstractions::data::{Data, Data::*};
use crate::error::ServerError;
use crate::protocol::type_schema::{TypeSchema, TypeSchema::*};
use crate::protocol::type_derive::*;
use crate::protocol::datatypes::SYSTEM_DATA_TYPE;
use crate::abstractions::portable_task::{TaskFunc, TaskPortableAbs, TaskPortableAbs::*, get_new_serial_id, zmqsend};
use crate::abstractions::fuse::{derivefuncpub, subfunc, coerce};

macro_rules! data_format_error {
    () => {
        Err(ServerError::DataErrorStr("internal data format error!"))
    };
}


pub fn fuse(a: &TaskPortableAbs, b: &TaskPortableAbs, force: bool) -> Option<TaskPortableAbs> {
    let fused_name = format!("({} -> {})", a.name(), b.name());
    if a.tout() == b.tin() {
        let a_func = a.sendfun();
        let b_func = b.sendfun();
        let new_serial_id = get_new_serial_id();
        let novel_func = Arc::new(move |input:Data, port: u16| {
            let result = a_func(input, port).and_then(|mid|b_func(mid, port));
            zmqsend(new_serial_id, port);
            result
        });
        let mut novel_signals = Vec::new();
        novel_signals.extend(a.signals().iter().cloned());
        novel_signals.push((a.serial_id(), a.name().clone()));
        novel_signals.extend(b.signals().iter().cloned());
        novel_signals.push((b.serial_id(), b.name().clone()));
        Some(Simple(fused_name, a.tin().clone(), b.tout().clone(), novel_func, new_serial_id, novel_signals))
    } else if let Some((substitution, construction)) = derivefuncpub(b.tin(), a.tout(), force, FuseDirection::Positive){
        let a_func = a.sendfun();
        let b_func = b.sendfun();
        let novel_output_type = construction.resolve(&b.tout());
        let new_serial_id = get_new_serial_id();
        let novel_func = Arc::new(move |input:Data, port: u16| {
            let temp = a_func.clone()(input, port)?;
            let temp = coerce(&substitution, temp)?;
            let result = destruct_apply(&construction, b_func.clone(), temp, port)?;
            zmqsend(new_serial_id, port);
            Ok(result)
        });
        let mut novel_signals = Vec::new();
        novel_signals.extend(a.signals().iter().cloned());
        novel_signals.push((a.serial_id(), a.name().clone()));
        novel_signals.extend(b.signals().iter().cloned());
        novel_signals.push((b.serial_id(), b.name().clone()));
        Some(Simple(fused_name, a.tin().clone(), novel_output_type, novel_func, new_serial_id, novel_signals))
    } else if let Some((substitution, construction)) = derivefuncpub(a.tout(), b.tin(), force, FuseDirection::Negative){
        let a_func = a.sendfun();
        let b_func = b.sendfun();
        let novel_input_type = construction.resolve(&a.tin());
        let new_serial_id = get_new_serial_id();
        let novel_func = Arc::new(move |input:Data, port: u16| {
            let temp = destruct_apply_coerce(&construction, a_func.clone(), &substitution, input, port)?;
            let result = b_func.clone()(temp, port)?;
            zmqsend(new_serial_id, port);
            Ok(result)
        });
        let mut novel_signals = Vec::new();
        novel_signals.extend(a.signals().iter().cloned());
        novel_signals.push((a.serial_id(), a.name().clone()));
        novel_signals.extend(b.signals().iter().cloned());
        novel_signals.push((b.serial_id(), b.name().clone()));
        Some(Simple(fused_name, novel_input_type, b.tout().clone(), novel_func, new_serial_id, novel_signals))
    } else {
        None
    }
}

type S = SubstitutionTransform;
type C = ConstructionTransform;

fn destruct_apply(c: &ConstructionTransform, f: TaskFunc, input: Data, port: u16) -> Result<Data, ServerError> {
    match c {
        C::Super => f(input, port),
        C::Void => Ok(input),
        C::Bool => Ok(input),
        C::Int => Ok(input),
        C::Float => Ok(input),
        C::Str => Ok(input),
        C::Json => Ok(input),
        C::DataType(name) => Ok(input),
        C::List(inner) => {
            match input {
                DataList(l) => {
                    let mut list = Vec::new();
                    for each in l {
                        list.push(destruct_apply(inner, f.clone(), each, port)?);
                    }
                    Ok(DataList(list))
                }
                _ => data_format_error!()
            }
        }
        C::Map(kk, vv) => {
            match input {
                DataMap(m) => {
                    let mut map = BTreeMap::new();
                    for (k, v) in m {
                        let k = destruct_apply(kk, f.clone(), k, port)?;
                        let v = destruct_apply(vv, f.clone(), v, port)?;
                        map.insert(k, v);
                    }
                    Ok(DataMap(map))
                }
                _ => data_format_error!()
            }
        }
        C::Tuple(ls) => {
            match input {
                DataTuple(d) => {
                    if ls.len() != d.len() {
                        data_format_error!()
                    } else {
                    let mut ahead = vec![];
                    let mut idx = 0;
                    for each in d {
                        ahead.push(destruct_apply(&ls[idx], f.clone(), each, port)?);
                            idx += 1;
                        }
                        Ok(DataTuple(ahead))
                    }
                }
                _ => data_format_error!()
            }
        }
        C::Opt(inner) => match input {
            DataVoid => Ok(DataVoid),
            _ => destruct_apply(inner, f, input, port)
        }
    }
}

fn destruct_apply_coerce(c: &ConstructionTransform, f: TaskFunc, s: &SubstitutionTransform, input: Data, port: u16) -> Result<Data, ServerError> {
    match c {
        C::Super => coerce(s, f(input, port)?),
        C::Void => Ok(input),
        C::Bool => Ok(input),
        C::Int => Ok(input),
        C::Float => Ok(input),
        C::Str => Ok(input),
        C::Json => Ok(input),
        C::DataType(name) => Ok(input),
        C::List(inner) => {
            match input {
                DataList(l) => {
                    let mut list = Vec::new();
                    for each in l {
                        list.push(destruct_apply_coerce(inner, f.clone(), s, each, port)?);
                    }
                    Ok(DataList(list))
                }
                _ => data_format_error!()
            }
        }
        C::Map(kk, vv) => {
            match input {
                DataMap(m) => {
                    let mut map = BTreeMap::new();
                    for (k, v) in m {
                        let k = destruct_apply_coerce(kk, f.clone(), s, k, port)?;
                        let v = destruct_apply_coerce(vv, f.clone(), s, v, port)?;
                        map.insert(k, v);
                    }
                    Ok(DataMap(map))
                }
                _ => data_format_error!()
            }
        }
        C::Tuple(ls) => {
            match input {
                DataTuple(d) => {
                    if ls.len() != d.len() {
                        data_format_error!()
                    } else {
                    let mut ahead = vec![];
                    let mut idx = 0;
                    for each in d {
                        ahead.push(destruct_apply_coerce(&ls[idx], f.clone(), s, each, port)?);
                            idx += 1;
                        }
                        Ok(DataTuple(ahead))
                    }
                }
                _ => data_format_error!()
            }
        }
        C::Opt(inner) => match input {
            DataVoid => Ok(DataVoid),
            _ => destruct_apply_coerce(inner, f, s, input, port)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::abstractions::data::Data;
    use crate::abstractions::data::Data::*;
    use crate::abstractions::task::TaskAbstraction;
    use crate::builtins::misc::BuiltinMisc;
    use crate::services::manager::init_service_manager;
    use crate::tasks::manager::init_task_manager;
    use crate::workflows::manager::init_workflow_manager;

    fn equivalent_operation(input: Vec<i64>) -> Vec<i64> {
        let mut result = Vec::new();
        for each in input {
            result.push((each + 1) * 2);
        }
        result
    }

    #[test]
    fn test_math_operation() {
        let add_1 = BuiltinMisc::list_add_1_portable();
        let mul_2 = BuiltinMisc::int_mul_2_portable();
        let fused = fuse(&add_1, &mul_2, false);
        let input = Data::DataList(vec![Data::DataInt(1), Data::DataInt(2), Data::DataInt(3)]);
        let output = fused.unwrap().sendfun()(input, 0).unwrap();
        let expected = Data::DataList(equivalent_operation(vec![1, 2, 3]).iter().map(|x| 
            Data::DataInt(*x)).collect()
        );
        assert_eq!(output, expected);
    }

    #[test]
    fn test_remote_tasks() {
        init_task_manager().unwrap();
        init_workflow_manager().unwrap();
        init_service_manager().unwrap();

        let task1 = TaskAbstraction::FromService(("http+--127.0.0.1+8080".to_string(), "list_add_1".to_string()));
        let task2 = TaskAbstraction::FromService(("http+--127.0.0.1+8080".to_string(), "int_mul_2".to_string()));
        let task1abs = TaskPortableAbs::from_taskabs(Arc::new(task1)).unwrap();
        let task2abs = TaskPortableAbs::from_taskabs(Arc::new(task2)).unwrap();
        let fused = fuse(&task1abs, &task2abs, false).unwrap();
        let input = Data::DataList(vec![Data::DataInt(1), Data::DataInt(2), Data::DataInt(3)]);
        let output = fused.sendfun()(input, 0).unwrap();
        let expected = Data::DataList(equivalent_operation(vec![1, 2, 3]).iter().map(|x| 
            Data::DataInt(*x)).collect()
        );
        assert_eq!(output, expected);
    }
}
